package cn.thu.thss.mla.logic;
import java.io.PrintWriter;
import java.util.HashMap;
import java.util.Iterator;

import cn.thu.thss.mla.structure.MessageInfo;
import cn.thu.thss.mla.structure.RowMutationInfo;


public class LogAnalyser {
	private HashMap<String, RowMutationInfo> logSet;
	
	private static final String CDR_NODE_ENQUEUE = "coordinatorNodeEnqueue";
	private static final String CDR_NODE_SEND = "coordinatorNodeSend";
	
	private static final String CDR_NODE_APPLY_ENQUEUE = "coordinatorNodeApplyEnqueue";
	private static final String CDR_NODE_APPLY_START = "coordinatorNodeApplyStart";
	private static final String CDR_NODE_APPLY_FINISH = "coordinatorNodeApplyFinish";
	
	private static final String SUB_NODE_RECEIVE = "subNodeReceive";
	private static final String SUB_NODE_ENQUEUE = "subNodeEnqueue";
	private static final String SUB_NODE_APPLY = "subNodeApply";
	private static final String SUB_NODE_FINISH = "subNodeFinish";
	
	
	public LogAnalyser() {
		logSet = new HashMap<String, RowMutationInfo>();
	}
	
	public void processLogLine(String line) {
		String[] segs = line.split("\t");
		String rmId = segs[0];
		RowMutationInfo rmInfo = logSet.containsKey(rmId) ? logSet.get(rmId) : new RowMutationInfo();
		rmInfo.setRowMutationId(rmId);	
		rmInfo.setKeyspaceName(segs[5]);
		rmInfo.setKeyInHex(segs[6]);
		
		Integer msgId = Integer.valueOf(segs[1]);
		HashMap<Integer, MessageInfo> messages = rmInfo.getMessages();
		MessageInfo msgInfo = messages.containsKey(msgId) ? msgInfo = messages.get(msgId) : new MessageInfo();
		msgInfo.setMessageId(msgId);

		String addr = segs[2];
		String logType = segs[3];
		long timeStamp = Long.valueOf(segs[4]);
		if (logType.equals(CDR_NODE_ENQUEUE)) {
			msgInfo.setSendTimeStart(timeStamp);
			msgInfo.setToAddress(addr);
		} else if (logType.equals(CDR_NODE_SEND)) {
			msgInfo.setSendTimeEnd(timeStamp);
			msgInfo.setToAddress(addr);
		} else if (logType.equals(SUB_NODE_RECEIVE)) {
			msgInfo.setRecvTimeStart(timeStamp);
			msgInfo.setFromAddress(addr);
		} else if (logType.equals(SUB_NODE_ENQUEUE)) {
			msgInfo.setEnqueueTimeStart(timeStamp);
			msgInfo.setFromAddress(addr);
		} else if (logType.equals(SUB_NODE_APPLY)) {
			msgInfo.setApplyTimeStart(timeStamp);
			msgInfo.setFromAddress(addr);
		} else if (logType.equals(SUB_NODE_FINISH)) {
			msgInfo.setApplyTimeFinish(timeStamp);
			msgInfo.setFromAddress(addr);
		} else if (logType.equals(CDR_NODE_APPLY_ENQUEUE)) {
			msgInfo.setEnqueueTimeStart(timeStamp);
			msgInfo.setFromAddress(addr);
			msgInfo.setToAddress(addr);
		} else if (logType.equals(CDR_NODE_APPLY_START)) {
			msgInfo.setApplyTimeStart(timeStamp);
			msgInfo.setFromAddress(addr);
			msgInfo.setToAddress(addr);
		} else if (logType.equals(CDR_NODE_APPLY_FINISH)) {
			msgInfo.setApplyTimeFinish(timeStamp);
			msgInfo.setFromAddress(addr);
			msgInfo.setToAddress(addr);
		} 
		messages.put(msgId, msgInfo);
		logSet.put(rmId, rmInfo);
	}
	
	public void printResults(PrintWriter good, PrintWriter bad) {
		Iterator<String> rmKeys = logSet.keySet().iterator();
		long count = 0;
		while (rmKeys.hasNext()) {
			RowMutationInfo rmInfo = logSet.get(rmKeys.next());
			Iterator<Integer> msgKeys = rmInfo.getMessages().keySet().iterator();
			while (msgKeys.hasNext()) {
				MessageInfo msgInfo = rmInfo.getMessages().get(msgKeys.next());
				msgInfo.calculateMetrics();
				count++;
				if (count % 10000 == 0) {
					System.out.println(count + " messages completed..");
				}
				PrintWriter out = msgInfo.getIsGoodMsgInfo() ? good : bad;
				out.print(rmInfo.getRowMutationId() + "\t");
				out.print(rmInfo.getKeyInHex() + "\t");
				out.print(msgInfo.getMessageId() + "\t");
				out.print(msgInfo.getFromAddress() + "\t");
				out.print(msgInfo.getToAddress() + "\t");
				out.print(msgInfo.getSendTimeInterval() + "\t");
				out.print(msgInfo.getRecvTimeInterval() + "\t");
				out.print(msgInfo.getEnqueueTimeInterval() + "\t");
				out.print(msgInfo.getApplyTimeInterval() + "\n");
			}
		}
		System.out.println(count + " messages completed totally");
	}
}
